/*
 * CPassiveThread.cpp
 *
 *  Created on: 2011-9-14
 *      Author: stargui
 */

#include "CPassiveThread.h"
#include <CommonLibrary.h>
#include <include/md5/md5.h>
#include <include/uuid/uuid.h>
#include "../SocketUtils/CPSelectUtils.h"
#include "../Protocol/ProtocolLabel.h"
#include "../Protocol/CProtocolFactory.h"
#include "CPassiveThread.h"
using namespace Common;
using namespace PROTOCOL;
using namespace Security;

CPassiveThread::CPassiveThread()
{
	_loads 			= 0;
	_release 			= false;
	_interval			= 0;
	_buffer 			= new char[4096];
	memset(_buffer,0,4096);
	_socketContainer.clear();
	_operation.clear();
	_clientNo			= 0;

	_operation[CERTIFICATE_CONNECT]			= boost::bind(&CPassiveThread::CertificateConnect, this, _1, _2, _3);
	_operation[CERTIFICATE_STREAM] 			= boost::bind(&CPassiveThread::CertificateStream, this, _1, _2, _3);
	_operation[CERTIFICATE_ALIVE]			= boost::bind(&CPassiveThread::CertificateKeepalive, this, _1, _2, _3);
	_operation[CERTIFICATE_DESCRIBE] 		= boost::bind(&CPassiveThread::CertificateDescribe, this, _1, _2, _3);
	_operation[CERTIFICATE_LOGIN] 			= boost::bind(&CPassiveThread::CertificateLogin, this, _1, _2, _3);
	_operation[CERTIFICATE_VERIFY] 			= boost::bind(&CPassiveThread::CertificateVerify, this, _1, _2, _3);
	_sessionManager 							= CSessionManager::GetInstance();
	//this->Start();//�������̺߳ϲ�
}

CPassiveThread::~CPassiveThread()
{
	if(_buffer)
	{
		delete[] _buffer;
		_buffer	= NULL;
	}
	_release		= true;
	ReleaseSockets();
	_operation.clear();
}

long CPassiveThread::GetCurrentLoads()
{
	return _loads;
}

void CPassiveThread::OnSocketConnect(int handle)
{
	_loads++;
	_socketContainer[handle].steps 		= 0;
	_socketContainer[handle].slave 		= false;
	_socketContainer[handle].interval 	= 0;
	_socketContainer[handle].taskindex	= _sessionManager->GetTaskIndex();
}

void CPassiveThread::SocketDisconnect(int handle)
{
	if (_socketContainer[handle].slave)
	{
		_sessionManager->RemoveSessionTask(_socketContainer[handle].session,_socketContainer[handle].taskindex);
	}
	else
	{
		_sessionManager->RemoveSession(handle, _socketContainer[handle].session);
	}
	_socketContainer.erase(handle);
	//��������ͷ�socket��֤ÿ�������е�socketΨһ
	CPSocketUtils::CloseSocket(handle);
}

void CPassiveThread::SocketInspectTimeout()
{
	if (GCL_ABS(time(NULL) - _interval) >= 5)
	{
		_interval = time(NULL);
		std::vector<int> dead;
		std::map<int, _TcpSocket>::iterator it;
		for (it = _socketContainer.begin(); it != _socketContainer.end(); it++)
		{
			if ((!it->second.slave) && (it->second.interval++ >= 5))
			{
				DEBUG_WARNING("------------socket:[%d]-------------\n",it->first);
				dead.push_back(it->first);
			}
		}
		std::vector<int>::iterator itr;
		for (itr = dead.begin(); itr != dead.end(); itr++)
		{
			SocketDisconnect(*itr);
		}
	}
}

void CPassiveThread::SetReleaseSockets()
{
	_release = true;
}

void CPassiveThread::ReleaseSockets()
{
	if (_release)
	{
		printf("-free-all--user--exit!!--\n");
		_release = false;
		std::map<int, _TcpSocket>::iterator it;
		int		sockfd				= -1;
		for (it = _socketContainer.begin(); it != _socketContainer.end(); it++)
		{
			sockfd					= it->first;
			CPSocketUtils::CloseSocket(sockfd);
		}
		_socketContainer.clear();
	}
}

void *CPassiveThread::ThreadFuncImpl()
{
	ReleaseSockets();
	if(_socketContainer.size() > 0)
	{
		CPSelectUtils selectutils;
		std::map<int, _TcpSocket>::iterator it;
		std::map<int, _TcpSocket>::iterator nextit;

		for (it = _socketContainer.begin(); it != _socketContainer.end(); )
		{
			if(!it->second.slave)
			{
				selectutils.InsertfdSets(it->first, P_FD_SETS_READ);
				selectutils.InsertfdSets(it->first, P_FD_SETS_ERROR);
			}

			if(ProcessTask(it->second.session,(int)it->second.slave,it->second.taskindex))
			{
				nextit				= it;
				nextit++;
				this->SocketDisconnect(it->first);
				it					= nextit;
			//	DEBUG_WARNING("---getpid:%d---size():%d------------------\n",getpid(),_socketContainer.size());
			}
			else
			{
				it++;
			}
		}

		if (selectutils.Select(100000) > 0)
		{
			for (it = _socketContainer.begin(); it != _socketContainer.end();)
			{
				if(it->second.slave)
				{
					it++;
					continue;
				}
				if (selectutils.CheckSets(it->first, P_FD_SETS_READ))
				{
					if (!this->SocketRecvTcpStream(it->first))
					{
						//DEBUG_WARNING("---link-center--failed!----getpid:[%d]--------\n",getpid());
						nextit					= it;
						nextit++;
						this->SocketDisconnect(it->first);
						it						= nextit;
					}
					else
					{
						it++;
						it->second.interval 	= 0;//ֻҪ����ȷ��ָ������Ϊ��������
					}
				}
				else if (selectutils.CheckSets(it->first, P_FD_SETS_ERROR))
				{
					nextit						= it;
					nextit++;
					this->SocketDisconnect(it->first);
					it							= nextit;
					//DEBUG_WARNING("---getpid:%d---size():%d------------------\n",getpid(),_socketContainer.size());
				}
				else
				{
					it++;
				}
			}
		}
		SocketInspectTimeout();
	}

	return NULL;
}
/**
 *  接受到数据时，跳转到的方法
 */
bool CPassiveThread::SocketRecvTcpStream(int handle)
{
 	try
 	{
 		int length = 0;
 		bool status = false;
 		//CProtocolFactory _packet;
 		std::map<int, _TcpSocket>::iterator _socket = _socketContainer.find(handle);
 		while ((length = CPSocketUtils::Recv(handle, _buffer, 4096, 0)) > 0)
 		{
 			status = true;
 			_socket->second.buffer.append(_buffer, length);
 		}

 		while ((length = _packet.Deserialize(_socket->second.buffer)) > 0)
 		{
 			//DEBUG_WARNING("--------------pt----------%d-----length:%d-------------------------",packet.PT(),length);
 			if (_packet.PT() == PAYLOAD_SIGNAL)
 			{
 				try
 				{
 					//Json::Value request;
 					//Json::Reader reader;
 					_reader.parse(_packet.PAYLOAD(), _request);

 					//DEBUG_WARNING("---handle:%d-getpid:%d-------%s----------\n",handle,getpid(),_packet.PAYLOAD().c_str());
 					std::string session = _request[SESSION].asString();
 					//根据模块处理不同的指令，MODULE_CERTIFICATE主要时登入指令和媒体注册指令
 					if (_request[MODULE] == MODULE_CERTIFICATE)
 					{
 						std::map<std::string, boost::function<bool(int, std::string &, Json::Value &)> >::iterator it;
 						if ((it = _operation.find(_request[OPERATION].asString())) != _operation.end())
 						{
 							if (!it->second(handle, session, _request))
 							{
 								return false;
 							}
 						}
 					}//登入成功后，改变状态
 					else if (_socket->second.steps == 0xFF)
 					{//处理其他的指令
 						if (!_sessionManager->HandleSessionRequest(handle, session, _request))
 						{
 							Json::Value resultResponse;
 							resultResponse[MODULE]			= _request[MODULE];
 							resultResponse[OPERATION]		= _request[OPERATION];
 							resultResponse[SESSION]			= _request[SESSION];
 							resultResponse[RESPONSE][ERRORCODE] 	= ANALYZEFAILED;
 							resultResponse[RESPONSE][ERRORCAUSE]	= EP_CN_ANALYZEFAILED;
 							if(!_request[PARAMETER][MEDIASTREAM_SSRC].isNull())
 							{
 								resultResponse[RESPONSE][MEDIASTREAM_SSRC]	= _request[PARAMETER][MEDIASTREAM_SSRC];
 							}
 							resultResponse[RESPONSE][MEDIASTREAM_STREAMNAME]	= _request[PARAMETER][MEDIASTREAM_STREAMNAME];
 							CProtocolFactory resultpacket;
 							std::string resultsender 				= resultpacket.Serialize(0, 0, resultResponse);
 							CPSocketUtils::Send(handle, resultsender.c_str(), resultsender.length(), 0,3000000,0);
 							return false;
 						}
 					}
 				}
 				catch (...)
 				{
 					DEBUG_WARNING("-----------%s----------\n",_packet.PAYLOAD().c_str());
 					DEBUG_WARNING("None Compatible Client Protocol!");
 					return false;
 				}
 			}
 			else if (_socket->second.slave)
 			{
 				_sessionManager->HandleSessionTask(handle, _socket->second.session, &_packet);
 			}
 			_socket->second.buffer.erase(0, length);
 		}
 		if (_socket->second.buffer.length() > 256 * 1024)
 		{
 			_socket->second.buffer.clear();
 		}
 		return status;
 	}
 	catch(...)
 	{
 		DEBUG_WARNING(" String None Compatible Client Protocol!");
		return false;
 	}
}

bool CPassiveThread::CertificateConnect(int handle, std::string &session, Json::Value &request)
{
	Json::Value response;
	if (_socketContainer[handle].steps == 0)
	{
		char uuid[64];
		Security::uuid_t out;
		Security::uuid_generate(out);
		Security::uuid_unparse(out, uuid);

		response[SESSION] 						= uuid;
		_socketContainer[handle].session 		= uuid;
		response[MODULE] 							= MODULE_CERTIFICATE;
		response[OPERATION] 						= request[OPERATION];

		Security::uuid_generate(out);
		Security::uuid_unparse(out, uuid);
		response[RESPONSE][CERTIFICATE_S0] 	= uuid;
		_socketContainer[handle].S0 = Security::hmac_md5(uuid, uuid);
		CProtocolFactory packet;
		std::string sender 						= packet.Serialize(0, 0, response);
		CPSocketUtils::Send(handle, sender.c_str(), sender.length(), 0,3000000,0);

		_socketContainer[handle].steps 			= 1;

		return true;
	}
	return false;
}

bool CPassiveThread::CertificateVerify(int handle, std::string &session, Json::Value &request)
{
	Json::Value response;
	if ((_socketContainer[handle].steps == 1) && (session == _socketContainer[handle].session))
	{
		if (_socketContainer[handle].S0 == request[PARAMETER][CERTIFICATE_S0].asString())
		{
			response[MODULE]			 				= MODULE_CERTIFICATE;
			response[SESSION] 						= session;
			response[OPERATION] 						= request[OPERATION];

			response[RESPONSE][CERTIFICATE_RETURN]	= true;
			response[RESPONSE][ERRORCODE] 			= SUCCESS_OK;
			response[RESPONSE][ERRORCAUSE]			= EP_CN_SUCCESS_OK;
			CProtocolFactory packet;
			std::string sender 						= packet.Serialize(0, 0, response);
			CPSocketUtils::Send(handle, sender.c_str(), sender.length(), 0,3000000,0);

			_socketContainer[handle].steps 			= 2;
			return true;
		}
		response[MODULE]			 					= MODULE_CERTIFICATE;
		response[SESSION] 							= session;
		response[OPERATION] 							= request[OPERATION];
		response[RESPONSE][CERTIFICATE_RETURN]		= false;
		response[RESPONSE][ERRORCODE] 				= SOFAILED;
		response[RESPONSE][ERRORCAUSE]				= EP_CN_SOFAILED;
		CProtocolFactory packet;
		std::string sender 							= packet.Serialize(0, 0, response);
		CPSocketUtils::Send(handle, sender.c_str(), sender.length(), 0,3000000,0);
	}
	return false;
}

bool CPassiveThread::CertificateLogin(int handle, std::string &session, Json::Value &request)
{
	Json::Value response;
	if ((_socketContainer[handle].steps == 2) && (session == request[SESSION].asString()))
	{
		response[SESSION] 								= session;
		response[MODULE] 									= MODULE_CERTIFICATE;
		response[OPERATION] 								= request[OPERATION];
		response[RESPONSE][CERTIFICATE_RETURN] 		= true;
		std::string user 									= "";
		int		loginflag									= 0;//0:Ĭ�ϣ�1��û���û����ֶΣ�����Ա��
		if(!request[PARAMETER][CERTIFICATE_USER].isNull())
		{
			if(!request[PARAMETER][CERTIFICATE_USER].asString().empty())
			{
				user										= request[PARAMETER][CERTIFICATE_USER].asString();
				loginflag									= 0;
			}
			else
			{
				loginflag									= 1;
			}
		}
		else
		{
			loginflag										= 1;
		}
		std::string password		 						= request[PARAMETER][CERTIFICATE_PASSWORD].asString();
		std::string mac									= "";
		if(!request[PARAMETER][CERTIFICATE_MAC].isNull())
		{
			mac												= request[PARAMETER][CERTIFICATE_MAC].asString();
		}

		int		iresult									= 0;
		int		level										= 3;
		if(!request[PARAMETER][CERTIFICATE_CID].isNull())
		{
			if(request[PARAMETER][CERTIFICATE_CID].isInt())
			{
				unsigned int	no								= request[PARAMETER][CERTIFICATE_CID].asUInt();
				if((_clientNo & 0x00FFFFFF) != (no & 0x00FFFFFF))
				{
					iresult									= -3;
				}
			}
		}

		if(!iresult)
		{
			iresult											= _sessionManager->SessionGetPermissions(user, password, level,mac,loginflag);
		}
		//		iresult			= 0;
		if(iresult)
		{
			response[RESPONSE][CERTIFICATE_RETURN] 		= false;
			switch(iresult)
			{
			case -1:
				response[RESPONSE][ERRORCODE] 					= NAMEORPASSWORDFAILED;
				response[RESPONSE][ERRORCAUSE]					= EP_CN_NAMEORPASSWORDFAILED;
				break;
			case -2:
				response[RESPONSE][ERRORCODE] 					= NN_USER_MAC_ILLEGAL;
				response[RESPONSE][ERRORCAUSE]					= EP_CN_NN_USER_MAC_ILLEGAL;
				break;
			case -3:
				response[RESPONSE][ERRORCODE] 					= NN_PERMISSION_DENIED;
				response[RESPONSE][ERRORCAUSE]					= EP_CN_NN_PERMISSION_DENIED;
				break;
			default:
				response[RESPONSE][ERRORCODE] 					= NAMEORPASSWORDFAILED;
				response[RESPONSE][ERRORCAUSE]					= EP_CN_NAMEORPASSWORDFAILED;
				break;
			}

			CProtocolFactory packet;
			std::string sender = packet.Serialize(0, 0, response);
			CPSocketUtils::Send(handle, sender.c_str(), sender.length(), 0,3000000,0);
			DEBUG_WARNING("--------------username-or-password-failed------------------\n");
			return false;
		}

//		response[RESPONSE][ERRORCODE] 			= SUCCESS_OK;
//		response[RESPONSE][ERRORCAUSE]			= EP_CN_SUCCESS_OK;

		if(!_sessionManager->InsertSession(handle, session,true,level,user,password))
		{//ʧ��
			response[RESPONSE][CERTIFICATE_RETURN] 		= false;
			response[RESPONSE][ERRORCODE] 					= TOOMANYUSERS;
			response[RESPONSE][ERRORCAUSE]					= EP_CN_TOOMANYUSERS;
			CProtocolFactory packet;
			std::string sender = packet.Serialize(0, 0, response);
			CPSocketUtils::Send(handle, sender.c_str(), sender.length(), 0,3000000,0);
			DEBUG_WARNING("--------------user--is-full----------------\n");
			return false;
		}

//		CProtocolFactory packet;
//		std::string sender 								= packet.Serialize(0, 0, response);
//		CPSocketUtils::Send(handle, sender.c_str(), sender.length(), 0,3000000,0);
		_socketContainer[handle].steps 					= 0xFF;
		_sessionManager->SendAlarmInfo(session);
		printf("--------client login success!!--------------\n");
		return true;
	}
	return false;
}

bool CPassiveThread::CertificateKeepalive(int handle, std::string &session, Json::Value &request)
{
	Json::Value response;
	if ((_socketContainer[handle].steps == 0xFF) && (session == request[SESSION].asString()))
	{
		response[SESSION]						= session;
		response[MODULE] 						= MODULE_CERTIFICATE;
		response[OPERATION] 					= request[OPERATION];

		CProtocolFactory packet;
		std::string sender 					= packet.Serialize(0, 0, response);
		if(CPSocketUtils::Send(handle, sender.c_str(), sender.length(), 0,3000000,0) < 0)
		{
			return false;
		}
		_socketContainer[handle].interval 	= 0;

		return true;
	}
	return false;
}

bool CPassiveThread::CertificateDescribe(int handle, std::string &session, Json::Value &request)
{
	Json::Value response;
	if ((_socketContainer[handle].steps == 0xFF) && (session == request[SESSION].asString()))
	{
		response[SESSION] 						    = session;
		response[MODULE] 							= MODULE_CERTIFICATE;
		response[OPERATION] 						= request[OPERATION];
		std::string describe;
		_sessionManager->SessionGetDescribe(describe);

		response[RESPONSE][CERTIFICATE_XML] 	= describe;
		CProtocolFactory packet;
		std::string sender = packet.Serialize(0, 0, response);
		CPSocketUtils::Send(handle, sender.c_str(), sender.length(), 0,3000000,0);
		return true;
	}
	return false;
}

bool CPassiveThread::CertificateStream(int handle, std::string &session, Json::Value &request)
{
	DEBUG_WARNING("the handle is %d,and streamname is %s :---getpid:[%d]-- \n",handle,request[PARAMETER][MEDIASTREAM_STREAMNAME].asString().c_str(),getpid());
	Json::Value response;
	CProtocolFactory packet;
	response[SESSION] 						= session;
	response[MODULE]				 			= MODULE_CERTIFICATE;
	response[OPERATION]						= request[OPERATION];
	int		iresult							= -1;

	iresult									= _sessionManager->InsertSessionTask(session, request[PARAMETER][MEDIASTREAM_STREAMNAME].asString(), handle,_socketContainer[handle].taskindex,this);
	switch(iresult)
	{
		case 0:
			{
				_socketContainer[handle].steps 		= 0xFF;
				_socketContainer[handle].slave 		= true;
				_socketContainer[handle].session 	= session;
				_socketContainer[handle].interval	= 0;
				response[RESPONSE][ERRORCODE]		= SUCCESS_OK;
				response[RESPONSE][ERRORCAUSE] 		= EP_CN_SUCCESS_OK;

			//	std::string sender 					= packet.Serialize(0, 0, response);
//				DEBUG_WARNING("--------------request success----------send ack------handle:[%d]----getpid:[%d]--------\n",handle,getpid());
			//	CPSocketUtils::Send(handle, sender.c_str(), sender.length(), 0,3000000,0);
				//return true;
			}
			break;
		case -1://media link register failed
			{
				response[RESPONSE][ERRORCODE]			= REGISTERMEDIASERVERFAILED;
				response[RESPONSE][ERRORCAUSE] 			= EP_CN_REGISTERMEDIASERVERFAILED;
				printf("--------------media link register fialed----------send ack------------------\n");
			}
			break;
		case -2:
			{
				response[RESPONSE][ERRORCODE]			= NN_MEDIA_LINK_EXIST;
				response[RESPONSE][ERRORCAUSE] 			= EP_CN_NN_MEDIA_LINK_EXIST;
				printf("--------------media link exist---------send ack------------------\n");
			}
			break;
		default:
			{
				response[RESPONSE][ERRORCODE]			= NORIGHT;
				response[RESPONSE][ERRORCAUSE] 			= EP_CN_NORIGHT;
				printf("--------------media link register no right----------send ack------------------\n");
			}
			break;
	}

	std::string sender 						= packet.Serialize(0, 0, response);
	CPSocketUtils::Send(handle, sender.c_str(), sender.length(), 0,3000000,0);

	return true;
}

//chech status by level 0:session,1:task 2:all
bool CPassiveThread::ProcessTask(std::string &session,int level,unsigned int taskindex)
{
	int status			= 0;
	_sessionManager->QuerySessionTask(session,level,taskindex,status);
	return status;
}


void CPassiveThread::SetClientNo(unsigned int no)
{
	_clientNo			= no;
}
